package com.mjf.market_analysis

import java.net.URL
import java.sql.Timestamp

import com.mjf.dim.{AdClickEvent, AdCountByProvince, BlackListWarning}
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * 广告点击量分析
 */
object AdAnalysisByProvince {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val resource: URL = getClass.getResource("/AdClickLog.csv")

    // 从文件读取数据
    val inputStream: DataStream[String] = env.readTextFile(resource.getPath)

    //
    val adLogStream: DataStream[AdClickEvent] = inputStream.map {
      line =>
        val dataArr: Array[String] = line.split(",")
        AdClickEvent(dataArr(0).toLong, dataArr(1).toLong, dataArr(2), dataArr(3), dataArr(4).toLong)
    }.assignAscendingTimestamps(_.timestamp * 1000L)

    // 定义刷单行为过滤操作
    val filterBlackListStream: DataStream[AdClickEvent] = adLogStream
      .keyBy(data => (data.userId, data.adId)) // 按照用户和广告ID分组
      .process(new FilterBlackList(100))

    val resultStream: DataStream[AdCountByProvince] = filterBlackListStream
      .keyBy(_.province)
      .timeWindow(Time.hours(1), Time.seconds(5))
      .aggregate(new AdCountAgg(), new AdCountResult())

    resultStream.print()
    filterBlackListStream.getSideOutput(new OutputTag[BlackListWarning]("blacklist")).print("blacklist")

    env.execute("AdAnalysisByProvince")

  }
}

// 判断用户对广告的点击次数是否达到上限
class FilterBlackList(maxClickCount: Int) extends KeyedProcessFunction[(Long, Long), AdClickEvent, AdClickEvent] {

  // 定义状态，需要保存点击量count
  lazy val countState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("count", classOf[Long]))

  // 标识位，用来表示用户是否已在黑名单中
  lazy val isSentSate: ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("is-sent", classOf[Boolean]))

  override def processElement(value: AdClickEvent, ctx: KeyedProcessFunction[(Long, Long), AdClickEvent, AdClickEvent]#Context, out: Collector[AdClickEvent]): Unit = {

    // 取出状态数据
    val curCount: Long = countState.value()

    // 如果是第一个数据，那么注册第二天0点的定时器，用于清空状态
    if (curCount == 0) {
      val ts: Long = (ctx.timerService().currentProcessingTime() / (1000 * 60 * 60 * 24) + 1) * (1000 * 60 * 60 * 24)
      ctx.timerService().registerProcessingTimeTimer(ts)
    }

    // 判断count值是否达到上限。如果达到，并且之前没有输出过报警信息，那么报警
    if (curCount >= maxClickCount) {
      if (!isSentSate.value()) {
        ctx.output(new OutputTag[BlackListWarning]("blacklist"), BlackListWarning(value.userId, value.adId, "click over" + maxClickCount + "times today"))
        isSentSate.update(true)
      }
      return
    }

    // count值加1
    countState.update(curCount + 1)

    out.collect(value)

  }

  // 0点触发定时器，直接清空状态
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[(Long, Long), AdClickEvent, AdClickEvent]#OnTimerContext, out: Collector[AdClickEvent]): Unit = {
    countState.clear()
    isSentSate.clear()
  }

}

class AdCountAgg() extends AggregateFunction[AdClickEvent, Long, Long] {
  override def createAccumulator(): Long = 0L

  override def add(value: AdClickEvent, accumulator: Long): Long = accumulator + 1

  override def getResult(accumulator: Long): Long = accumulator

  override def merge(a: Long, b: Long): Long = a + b
}

class AdCountResult() extends WindowFunction[Long, AdCountByProvince, String, TimeWindow] {
  override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[AdCountByProvince]): Unit = {

    out.collect(AdCountByProvince(key, new Timestamp(window.getEnd).toString, input.head))

  }
}